package cn.middle.remote.server;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.middle.remote.Buffer;
import cn.middle.remote.enums.CallState;
import cn.middle.remote.message.AckSignal;
import cn.middle.remote.message.BaseSignal;
import cn.middle.remote.message.CallSignal;
import cn.middle.remote.message.EdgeReplySignal;
import cn.middle.remote.message.EdgeSignal;
import cn.middle.remote.message.EventReportSignal;
import cn.middle.remote.message.HeartbeatPingSignal;
import cn.middle.remote.message.HeartbeatPongSignal;
import cn.middle.remote.message.LoginResultSignal;
import cn.middle.remote.message.LoginSignal;
import cn.middle.remote.message.RegistryEdgeSignal;
import cn.middle.remote.message.ReplyMessageSignal;
import cn.middle.remote.message.SignalHelper;
import cn.middle.remote.message.SubscribeSignal;
import cn.middle.remote.util.NettyAttributeUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;

public class NettyServerProcessHandler extends ServerProcessingHandler {
	
	private final Logger log = LoggerFactory.getLogger(NettyServerProcessHandler.class);
	
	ExecutorService executors = Executors.newFixedThreadPool(10);

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		log.info("remoteAddress {} to Connection",ctx.channel().remoteAddress());
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		Integer clientId = NettyAttributeUtil.clientId(ctx.channel());
		this.listener.connectLost(clientId);
		if(clientId == null) {
			return ;
		}
		Session session = getClientId(clientId);
		if (session != null) {
			session.close();
			CLIENTS.remove(clientId);
		}
		
	}

	@Override
	public void received(ChannelHandlerContext ctx, BaseSignal msg) throws Exception {
		byte messageType = msg.getMessageType();
		if(ifLogin(msg)) {
			loginHandler(ctx, (LoginSignal)msg);
			return ;
		}
		if(hasAuth(ctx.channel())) {//有权限才进入处理
			executors.execute(()->{
				switch(messageType) {
				case SignalHelper.CALL:
					callForward(ctx, (CallSignal)msg);
					break;
				case SignalHelper.REPLY:
					replyForward(ctx, (ReplyMessageSignal)msg);
					break;
				case SignalHelper.EVENT_REPORT:
					event(ctx, (EventReportSignal)msg);
					break;
				case SignalHelper.SUBSCRIBE:
					subscribe(ctx, (SubscribeSignal)msg);
					break;
				case SignalHelper.EDGE:
					edge(ctx, (EdgeSignal)msg);
					break;
				case SignalHelper.REGISTRY_EDGE:
					registry(ctx, (RegistryEdgeSignal)msg);
					break;
				case SignalHelper.PING:
					ping(ctx, (HeartbeatPingSignal)msg);
					break;
				case SignalHelper.ACK:
					ack(ctx,(AckSignal)msg);
					break;
				}
			});
		}
	}
	
	private void ack(ChannelHandlerContext ctx,AckSignal msg) {
		listener.ack(NettyAttributeUtil.clientId(ctx.channel()), msg.getId());
	}
	
	private void subscribe(ChannelHandlerContext ctx,SubscribeSignal msg) {
		sendAck(ctx, msg.getId());
		listener.subscribe(NettyAttributeUtil.clientId(ctx.channel()), new Buffer(msg.getId(),msg.getData(),getSerialization()));
	}
	
	private void registry(ChannelHandlerContext ctx,RegistryEdgeSignal msg) {
		sendAck(ctx, msg.getId());
		listener.registryEdge(NettyAttributeUtil.clientId(ctx.channel()), new Buffer(msg.getId(),msg.getData(),getSerialization()));
	}
	
	private void event(ChannelHandlerContext ctx,EventReportSignal msg) {
		sendAck(ctx, msg.getId());
		listener.event((clientIds,obj) ->{
			if(clientIds != null && !clientIds.isEmpty()) {
				if(obj != null) {
					msg.setData(serialization.encode(obj));
				}
				for(Integer clientId : clientIds) { //符合条件的集合
					Session session = getClientId(clientId);
					if(session  != null) {
						//直接推送，无需处理
						session.send(msg);
					}
				}
			}
		}, msg.getEventId(),new Buffer(msg.getId(),msg.getData(),getSerialization()) );
	}
	
	
	private void sendAck(ChannelHandlerContext ctx,String msgId) {
		Integer clietnId = NettyAttributeUtil.clientId(ctx.channel());
		Session session = getClientId(clietnId);
		session.send(new AckSignal(msgId));
	}
	
	
	private void edge(ChannelHandlerContext ctx,EdgeSignal msg) {
		listener.edge(obj -> {
			int clientId = NettyAttributeUtil.clientId(ctx.channel());
			Session session = getClientId(clientId);
			if(session != null) {
				EdgeReplySignal signal = new EdgeReplySignal();
				signal.setId(msg.getId());
				signal.setData(serialization.encode(obj));
				session.send(signal);
			}
		}, new Buffer(msg.getId(),msg.getData(),getSerialization()));
	}
	/**
	 *心跳回复
	 * @param ctx
	 * @param signal
	 */
	private void ping(ChannelHandlerContext ctx,HeartbeatPingSignal signal) {
		log.info("the client {} ping",NettyAttributeUtil.clientId(ctx.channel()));
		ctx.channel().writeAndFlush(new HeartbeatPongSignal(signal.getId()));
	}
	
	/**
	 * 鉴权
	 * @param ctx
	 * @param msg
	 */
	private void loginHandler(ChannelHandlerContext ctx,LoginSignal msg) {
		Integer clientId = msg.getClientId();
		String username = msg.getUsername();
		String password = msg.getPassword();
		LoginValidation validation = authHandler.auth(username, password, clientId);
		if(validation.isSuccess()) { //校验成功
			// 鉴权成功！
			NettyAttributeUtil.setLogin(ctx.channel(), true);
			NettyAttributeUtil.username(ctx.channel(), username);
			NettyAttributeUtil.clientId(ctx.channel(), clientId);
			LoginResultSignal signal = new LoginResultSignal();
			signal.setCode(0);
			signal.setMessage(validation.getLoginMessage());
			ctx.writeAndFlush(signal);
			//ctx.channel().writeAndFlush(signal);
			CLIENTS.put(clientId,new NettySession(ctx.channel()));//存入缓存中
			listener.connectAfter(clientId);
		    listener.login(username, password, clientId);
		}else {
			// 鉴权失败！
			LoginResultSignal signal = new LoginResultSignal();
			signal.setCode(-1);
			signal.setMessage(validation.getLoginMessage());
			ctx.writeAndFlush(signal);
			ctx.close(); //关闭
		}
	}
	
	/**
	 * 判断是否有权限
	 * @param channel 通道
	 * @return
	 */
	public boolean hasAuth(Channel channel) {
		return NettyAttributeUtil.isLogin(channel);
	}
	
	/**
	 * 是否是登陆请求
	 * 
	 * @param msg
	 * @return
	 */
	private boolean ifLogin(BaseSignal msg) {
		return SignalHelper.LOGIN == msg.getMessageType();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		log.error(cause.getMessage() + " : "+ NettyAttributeUtil.clientId(ctx.channel()),cause);
		this.listener.exception(NettyAttributeUtil.clientId(ctx.channel()), cause);
	}

	@Override
	public void clearUp() {
		for(Session session : this.CLIENTS.values()) {
			session.close();
		}
	}
	
	/**
	 * 调用转发
	 * @param signal
	 */
	private void callForward(ChannelHandlerContext ctx,CallSignal signal) {
		int remoteClientId = signal.getRemoteAddressClientId();
		int sendClientId = NettyAttributeUtil.clientId(ctx.channel());
		if(CLIENTS.containsKey(remoteClientId)) {
			Session session = CLIENTS.get(remoteClientId);
			signal.setRemoteAddressClientId(sendClientId); //将目的标识更改为发送的地址（边缘端执行完成后按这个地址发送回来）
			session.send(signal); //发送
			listener.call(sendClientId, remoteClientId, signal.getId(), signal.getInterfaceName(), new Buffer(signal.getId(),signal.getData(),this.serialization));
		}else {
			ReplyMessageSignal reply = new ReplyMessageSignal();
			reply.setCode(CallState.FAIL.getCode()); //没有查找到表示还没有在线或者没有需要管理者检查。
			reply.setMessage("not find the remote address in online");
			reply.setId(signal.getId());
			reply.setRemoteAddressClientId(sendClientId); 
			reply.setInterfaceName(signal.getInterfaceName());
			ctx.writeAndFlush(reply);
		}
	}
	
	/**
	 * 边缘端回复转发
	 * @param signal
	 */
	private void replyForward(ChannelHandlerContext ctx,ReplyMessageSignal signal) {
		int remoteClientId = signal.getRemoteAddressClientId();
		int sendClientId = NettyAttributeUtil.clientId(ctx.channel());
		if(CLIENTS.containsKey(remoteClientId)) {
			Session session = CLIENTS.get(remoteClientId);
			session.send(signal); //发送
			this.listener.reply(sendClientId, remoteClientId, signal.getId(), signal.getInterfaceName(), new Buffer(signal.getId(),signal.getData(),serialization));
			return ;
		}
		log.info("the messageId {} reply is fail,because the client {} not online",signal.getId(),remoteClientId);
	}

}
